package com.zallds.window;

import com.zallds.utils.Utils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 全窗口函数
 * ProcessWindowFunction
 * 需求：每经过3秒钟，求安放在北京西站各个红外测温仪所探测到的所有旅客的平均体温。
 */
public class FullWindowFunctionDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.62.37", 8889)
                .flatMap(new Utils.Splitter())
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
                .process(new MyProcessWindowFunction())
                .print();

        env.execute();
    }


    /**
     *   自定义的ProcessWindowFunction抽象类的子类型
     *   
     *   @tparam IN The type of the input value. → 待处理的DataStream中每个元素的类型，此处是：Tuple2
     *   @tparam OUT The type of the output value. → 结果DataStream中每个元素的类型，此处是：(String,Double)
     *   @tparam KEY The type of the key. → keyBy中指定的key的类型，此处是String
     *   @tparam W The type of the window. → 窗口的类型，此处是TimeWindow
     *   
     */
    private static class MyProcessWindowFunction extends ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Long>, String, TimeWindow> {

        /**
         * Evaluates the window and outputs none or several elements
         * 窗口关闭时，process方法会触发执行
         *
         * @param name      分组的key字段
         * @param context   待求值的窗口上下文的实例
         * @param elements  窗口中待计算的元素
         * @param out 用于将计算后的结果送往目标DataStream中存储起来
         * @throws Exception
         */
        @Override
        public void process(String name, Context context, Iterable<Tuple2<String, Integer>> elements, Collector<Tuple2<String, Long>> out) throws Exception {
            int cnt = 0;
            Long sum = 0L;

            for (Tuple2<String, Integer> tuple : elements) {
                cnt++;
                sum = sum + tuple.f1;
            }
            out.collect(new Tuple2<>(name, sum / cnt));
        }
    }
}
